package es

import (
	"context"
	"fmt"
	"github.com/MingOf/logDemo/kafka"
	"github.com/Shopify/sarama"
	"github.com/olivere/elastic/v7"
)

type Es struct {
	CurrentIndex string
	*elastic.Client
}

/*
创建 elastic client
*/
func NewEsClient(address string) *Es {
	cli, err := elastic.NewClient(elastic.SetURL("http://" + address))
	if err != nil {
		panic(err)
	}
	fmt.Println("connect to es success")
	return &Es{
		"",
		cli,
	}
}

//得到的值并非按照理想中的排序，这是因为 kafka 只能保证每个分区的顺序，并不能保证分区之间的日志的顺序。
//并且写入es的操作也是异步的，所以写入es的顺序是无法预测的
func (e *Es) SendToEs(pc sarama.PartitionConsumer) {
	messages := pc.Messages()
	for m := range messages {
		data := kafka.Data{
			Partition: m.Partition,
			Offset:    m.Offset,
			Key:       string(m.Key),
			Value:     string(m.Value),
		}
		// elastic index操作，将数据写入index
		ir, err := e.Index().
			Index(e.CurrentIndex).
			BodyJson(data).
			Do(context.Background())
		if err != nil {
			panic(err)
		}
		fmt.Printf("Indexed student %s to index %s, type %s, data:%#v\n", ir.Id, ir.Index, ir.Type, data)
	}
	defer pc.AsyncClose()
}
